package com.gitee.xhs.window.evictors;

import com.gitee.xhs.OperatorSourceFactory;
import com.gitee.xhs.customFunction.OperatorKeyBy;
import com.gitee.xhs.customWaterMark.CustomWaterStrategy;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import com.gitee.xhs.window.windowOperator.DefaultAggregateWindowFunction;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author BigHead
 * @date 2022/7/27 9:46
 * @description SimpleEvictors
 */
public class SimpleEvictors {

    private final StreamExecutionEnvironment streamExecutionEnvironment;

    public SimpleEvictors(StreamExecutionEnvironment streamExecutionEnvironment) {
        this.streamExecutionEnvironment = streamExecutionEnvironment;
    }

    public void evictorsLogic() {

        WindowedStream<Transaction, Long, TimeWindow> windowedStream = streamExecutionEnvironment
                .addSource(
                        OperatorSourceFactory.unBoundLoopOneToTenOneSecondSkipSource()
                )
                .assignTimestampsAndWatermarks(
                        new CustomWaterStrategy()
                )
                .keyBy(
                        new OperatorKeyBy()
                )
                .window(
                        TumblingEventTimeWindows.of(Time.minutes(1))
                );

        // evictors before
        beforeEvictors(windowedStream);
        // evictors after
        afterEvictors(windowedStream);

    }

    private void beforeEvictors(WindowedStream<Transaction, Long, TimeWindow> windowedStream) {
        windowedStream
                .evictor(
                        new CustomBeforeEvictors()
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("Evictors Before 设定保留一条数据效果");
    }

    private void afterEvictors(WindowedStream<Transaction, Long, TimeWindow> windowedStream) {
        windowedStream
                .evictor(
                        new CustomAfterEvictors()
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("Evictors After 设定保留2条数据效果");
    }


}
